package com.aqs.demo;

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 使用Condition实现一个有界队列
 * 调用add方法时，如果元素达到上限，让当前线程等待
 * 移除元素时，调用Condition中的signal()方法，将等待队列的头结点唤醒。（节点类型与AQS的节点类型是一样的）
 *
 * @author xudong
 * @date 2024/02/16
 */
public class BoundedQueue<T> {

	private final Object[] arrays;

	/** 添加元素时的索引 */
	private int addIndex;
	/** 移除元素时的索引 */
	private int removeIndex;
	/** 元素个数 */
	private int count;

	private final ReentrantLock lock = new ReentrantLock();
	private final Condition notEmpty = lock.newCondition();
	private final Condition notFull = lock.newCondition();


	public BoundedQueue(int capacity) {
		arrays = new Object[capacity];
	}

	public void add(T t) {
		lock.lock();
		try {
			while (count == arrays.length) {
				notFull.await();
			}
			System.out.println("线程 " + Thread.currentThread().getName() + " add " + t);
			arrays[addIndex] = t;
			if (++addIndex == arrays.length) {
				addIndex = 0;
			}
			count++;
			notEmpty.signal();
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		} finally {
			lock.unlock();
		}

	}

	public T remove() {
		lock.lock();
		System.out.println("线程 " + Thread.currentThread().getName() + " 进入remove ");
		try {
			while (count == 0) {
				System.out.println("没有元素，线程阻塞 " + Thread.currentThread().getName());
				notEmpty.await();
				System.out.println("线程释放 " + Thread.currentThread().getName());
			}
			T t = (T) arrays[removeIndex];
			System.out.println("线程 " + Thread.currentThread().getName() + " remove success " + t);
			if (++removeIndex == arrays.length - 1) {
				removeIndex = 0;
			}
			count--;
			// 从阻塞队列头部唤醒一个线程
			notFull.signal();
			return t;
		} catch (InterruptedException e) {
			throw new RuntimeException(e);
		} finally {
			lock.unlock();
		}
	}

	public static void main(String[] args) {
		final int max = 20;
		BoundedQueue<Integer> boundedQueue = new BoundedQueue<>(max);

		for (int i = 10; i < max; i++) {
			new Thread(() -> {
				boundedQueue.remove();
			}, i + "").start();
		}

		for (int i = 0; i < max; i++) {
			int finalI = i;
			new Thread(() -> {
				boundedQueue.add(finalI);
			}, i + "").start();
		}
	}
}


